from sqlalchemy import Column, Integer, String
from sqlalchemy.orm.session import Session
import os
import logging
from airflow.models.base import Base
from airflow.utils.session import provide_session

log = logging.getLogger(__name__)


class Config(Base):
    __tablename__ = "idata_config"
    __table_args__ = {"useexisting": True}

    id = Column(Integer, primary_key=True, autoincrement=True)
    section = Column(String(128))
    key = Column(String(128))
    value = Column(String(255))

    @staticmethod
    @provide_session
    def get(section, key, session: Session = None):
        return session.query(Config).filter(Config.section == section, Config.key == key).first()

    @staticmethod
    @provide_session
    def find_all(session: Session = None):
        return session.query(Config).all()

    @staticmethod
    @provide_session
    def get_by_id(config_id, session: Session = None):
        return session.query(Config).filter(Config.id == config_id).first()

    @staticmethod
    @provide_session
    def query_page(page, section=None, session: Session = None, size=10):
        q = session.query(Config).order_by(Config.section)
        if section:
            q = q.filter(Config.section == section)
        return q.paginate(page, size, error_out=False)

    @provide_session
    def insert(self, session: Session = None):
        name = env_var_name(self.section, self.key)
        os.environ[name] = self.value
        return session.add(self)

    @staticmethod
    @provide_session
    def delete(config_id, session: Session = None):
        session.query(Config).filter(Config.id == config_id).delete()

    @provide_session
    def update(self, session: Session = None):
        name = env_var_name(self.section, self.key)
        os.environ[name] = self.value
        session.merge(self)

    def to_json(self):
        if self is None:
            return None
        data = {
            'id': self.id,
            'section': self.section,
            'key': self.key,
            'value': self.value,
        }
        return data


def env_var_name(section, key):
    return f'AIRFLOW__{section.upper()}__{key.upper()}'


def init_env_config():
    log.info('init os env from db ...')
    for config in Config.find_all():
        try:
            name = env_var_name(config.section, config.key)
            os.environ[name] = config.value
            log.info('{} = {}'.format(name, config.value))
        except Exception as e:
            log.exception(e)
